package cn.westv.scps.mqtt;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.json.simple.parser.ParseException;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xiaomi.xmpush.server.Constants;
import com.xiaomi.xmpush.server.Message;
import com.xiaomi.xmpush.server.Result;
import com.yunpian.sdk.model.SmsBatchSend;

import MIPush.ExecuteMIPush;
import MIPush.MessageInfo;
import MIPush.messagefactory.MessageFactory;
import cn.westv.scps.connTools.ProxyConnExec;
import cn.westv.scps.domain.BikeOwnerEntity;
import cn.westv.scps.domain.SmsTemplateEntity;
import cn.westv.scps.domain.SysMessageEntity;
import cn.westv.scps.service.bikeowner.IBikeOwnerService;
import cn.westv.scps.service.sms.template.ISmsTemplateService;
import cn.westv.scps.service.sys.config.ISysConfigService;
import cn.westv.scps.service.sysmsg.ISysMsgService;
import cn.westv.scps.utils.DateUtil;
import cn.westv.scps.utils.MD5Util;
import cn.westv.scps.utils.PropertiesUtil;
import cn.westv.scps.utils.SpringContextUtil;
import cn.westv.scps.utils.StringUtils;
import cn.westv.scps.web.utils.SendSmsTools;

/**
 * 发布消息的回调类
 * 
 * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
 * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。 在回调中，将它用来标识已经启动了该回调的哪个实例。
 * 必须在回调类中实现三个方法：
 * 
 * public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
 * 
 * public void connectionLost(Throwable cause)在断开连接时调用。
 * 
 * public void deliveryComplete(MqttDeliveryToken token)) 接收到已经发布的 QoS 1 或 QoS 2
 * 消息的传递令牌时调用。 由 MqttClient.connect 激活此回调。
 * 
 */
public class PushCallback implements MqttCallback {
	private MqttClient client;
	private Logger log;
	public static Set<String> mqttMsgSet = new HashSet<>();// 保存这当天接收过得需要处理的mqtt主题

	private static final String abnormitySmsTipc = "message/abnormity/sms/.+";
	private static final String smsTipc = "message/.+/sms/.+";
	private static final String appTipc = "message/.+/app/.+";
	private static final String wechatTipc = "message/.+/wechat/.+";
	private static final String ownerRemoveTipc = "sysbiz/owner/remove/.+";

	private ISysConfigService sysConfigService;

	public PushCallback(MqttClient client) {
		log = Logger.getLogger(PushCallback.class);
		this.client = client;
	}

	public void connectionLost(Throwable cause) {
		MqttConnectOptions options = new MqttConnectOptions();
		String username = PropertiesUtil.getInstance().getPro("", "mqtt.username");
		String password = PropertiesUtil.getInstance().getPro("", "mqtt.password");
		// MQTT的连接设置
		// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
		options.setCleanSession(true);
		// 设置连接的用户名
		options.setUserName(username);
		// 设置连接的密码
		options.setPassword(password.toCharArray());
		// 设置超时时间 单位为秒
		options.setConnectionTimeout(10);
		// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
		options.setKeepAliveInterval(20);
		boolean falg = true;
		while (falg) {
			try {
				client.connect(options);
				ClientMQTT.subscribe(client);
				falg = false;
				log.info("连接断开，重连........成功！");
			} catch (MqttException e1) {
				log.info("mqtt订阅失败！");
			} catch (Exception e) {
				// e.printStackTrace();
				log.info("mqtt连接失败！");
			}
			if(falg){
				try {
					Thread.sleep(5);// 等待5秒以后，继续尝试连接
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		
	}

	public void deliveryComplete(IMqttDeliveryToken token) {
		log.debug("deliveryComplete---------" + token.isComplete());
	}

	public void messageArrived(String topic, MqttMessage message) throws Exception {
		try {
			String msgJSONstr = URLDecoder.decode(new String(message.getPayload()), "UTF-8");
			log.debug("接收消息主题 : " + topic + ";接收消息Qos : " + message.getQos() + ";接收消息内容 : " + msgJSONstr);
			// subscribe后得到的消息会执行到这里面
			// 这里的判断，订阅的是所有的主题内容，去重复处理就放在需要处理的信息后面（即，只有需要处理的信息，才进行去重复处理）所以判断条件是
			// topic.matches("message/.+/app/.+")&&repetitionFilter(topic,message)，下面的判断也如此
			if (topic.matches(appTipc) && repetitionFilter(topic, message)) {// 推送消息
				hendleMIPush(topic, message);// 小米推送
			}
			if (topic.matches(smsTipc) && repetitionFilter(topic, message)) {// 发送短信消息
				hendleSmsMessage(topic, message);
			}
			if (topic.matches(wechatTipc) && repetitionFilter(topic, message)) {// 微信推送
				hendleWechatSend(topic, message);
			}
			if (topic.matches(ownerRemoveTipc) && repetitionFilter(topic, message)) {// base服务器注销用户
				log.info("接收消息主题 : " + topic + ";接收消息Qos : " + message.getQos() + ";接收消息内容 : " + msgJSONstr);
				IBikeOwnerService bikeOwnerServiceImpl = (IBikeOwnerService) SpringContextUtil
						.getBean("bikeOwnerServiceImpl");
				JSONObject jsonObject = JSON.parseObject(msgJSONstr);
				String mobilesStr = jsonObject.getString("mobiles");
				String[] mobilesArr = mobilesStr.split(",");
				for (String mobile : mobilesArr) {
					BikeOwnerEntity owner = bikeOwnerServiceImpl.findByMobile(mobile);
					bikeOwnerServiceImpl.deleteById(BikeOwnerEntity.class, owner.getBikeOwnerId());
				}
			}

		} catch (Exception e) {
			ByteArrayOutputStream out = null;
			String string = "";
			try {
				out = new ByteArrayOutputStream();
				PrintStream s = new PrintStream(out);
				e.printStackTrace(s);
				string = new String(out.toByteArray());
			} catch (Exception e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			} finally {
				if (out != null)
					out.close();
			}
			log.error(string);
			log.debug(string);
		}
	}

	/**
	 * 微信推送
	 * 
	 * @param topic
	 * @param message
	 * @throws Exception
	 */
	private void hendleWechatSend(String topic, MqttMessage message) throws Exception {
		String msgJSONstr = URLDecoder.decode(new String(message.getPayload()), "UTF-8");
		log.info("接收消息主题 : " + topic + ";接收消息Qos : " + message.getQos() + ";接收消息内容 : " + msgJSONstr);
		JSONObject jsonObject = JSON.parseObject(msgJSONstr);
		String openid = topic.substring(topic.lastIndexOf("/") + 1);
		String msg = jsonObject.getString("msg");
		JSONObject sendWechatTxtMsg = ProxyConnExec.sendWechatTxtMsg(openid, msg);
		int i = 3;
		while ((sendWechatTxtMsg == null || sendWechatTxtMsg.getString("opStatus").equals("1"))) {

			sendWechatTxtMsg = ProxyConnExec.sendWechatTxtMsg(openid, msg);
			log.info("发送微信失败，重试！");
			i--;
			if (i < 0) {
				return;// 如果重试3此后仍然不能成功，则停止重试！
			}
		}
		log.info("发送微信成功！");
	}

	/**
	 * 推送消息的处理，小米推送
	 * 
	 * @param topic
	 * @param message
	 * @throws UnsupportedEncodingException
	 * @throws IOException
	 * @throws ParseException
	 */
	private void hendleMIPush(String topic, MqttMessage message)
			throws UnsupportedEncodingException, IOException, ParseException {
		String msgJSONstr = URLDecoder.decode(new String(message.getPayload()), "UTF-8");
		log.info("接收消息主题 : " + topic + ";接收消息Qos : " + message.getQos() + ";接收消息内容 : " + msgJSONstr);
		JSONObject jsonObject = JSON.parseObject(msgJSONstr);
		String mobile = topic.substring(topic.lastIndexOf("/") + 1);
		String code = jsonObject.getString("msgCode");
		String time = jsonObject.getString("time");
		ISysMsgService sysMsgService = (ISysMsgService) SpringContextUtil.getBean("sysMsgServiceImpl");
		SysMessageEntity messageEntity = sysMsgService.findByCode(code);
		String payload = messageEntity.getPayload();
		String[] timeSplit = time.split(" ");
		time = timeSplit[1];
		payload = payload.replace("#time#", time);
		messageEntity.setPayload(payload);
		pushMessage(mobile, messageEntity);// 执行小米推送逻辑
	}

	/**
	 * 短信发送消息处理，云片短信
	 * 
	 * @param topic
	 * @param message
	 * @throws UnsupportedEncodingException
	 */
	private void hendleSmsMessage(String topic, MqttMessage message) throws UnsupportedEncodingException {
		String msgJSONstr = URLDecoder.decode(new String(message.getPayload()), "UTF-8");
		log.info("接收消息主题 : " + topic + ";接收消息Qos : " + message.getQos() + ";接收消息内容 : " + msgJSONstr);
		JSONObject jsonObject = JSON.parseObject(msgJSONstr);
		String mobile = topic.substring(topic.lastIndexOf("/") + 1);
		String code = jsonObject.getString("msgCode");
		if (topic.matches(abnormitySmsTipc)) {// 告警类消息
			String type = jsonObject.getString("type");
			Map<String, Integer> countMap = hendleMobileSendAlarmCount(mobile, type, "get");
			int surplusCount = 0;
			surplusCount = surplusCount(type, countMap.get(type));
			if (surplusCount > 0) {// 告警类消息是否超过限制数量,大于0表示允许发送，小于或等于0表示不允许发送
				// 发送短信，这里的6-(countMap.get("total")+1)表达的是还剩余发送短信次数，因为countMap.get("total")得到的数量是发送以前的数量，为了短信内容的正确，所以需要进行+1处理
				Map<String, Object> sendSms = sendSms(jsonObject, mobile, code, 6 - (countMap.get("total") + 1));
				com.yunpian.sdk.model.Result<SmsBatchSend> result = (com.yunpian.sdk.model.Result<SmsBatchSend>) sendSms
						.get("result");
				if (result.getCode() == 0 && result.getData().getData().get(0).getCode() == 0) {
					hendleMobileSendAlarmCount(mobile, type, "add");// 告警短信次数添加
				}
			} else {
				log.info(mobile + "：这个手机号今天发送告警消息超过限制，所以不给予发送短信！");
			}
		} else {
			sendSms(jsonObject, mobile, code, 0);// 不属于告警类消息的，直接发送短信
		}
	}

	/**
	 * 获取告警次数,或者增加告警次数
	 * 
	 * @param mobile
	 * @param way
	 *            "add"表示增加告警次数,"get"代表获取次数
	 * @return
	 */
	private synchronized Map<String, Integer> hendleMobileSendAlarmCount(String mobile, String type, String way) {
		Map<String, Map<String, Integer>> map = SendSmsTools.mobileSendAlarmCountMap;
		Map<String, Integer> mobileMap = map.get(mobile);

		if (mobileMap == null) {
			mobileMap = new HashMap<>();
			mobileMap.put(type, 0);
			map.put(mobile, mobileMap);
		}
		Integer count = mobileMap.get(type);
		if (count == null) {
			count = 0;
			mobileMap.put(type, count);
		}
		if (way.equals("get")) {// 返回数量即可
			if (!mobileMap.containsKey("total")) {
				mobileMap.put("total", 0);
			}
			return mobileMap;
		}
		mobileMap.remove("total");// 先清除总数两
		Set<Entry<String, Integer>> entrySet = mobileMap.entrySet();
		int total = 0;
		for (Entry<String, Integer> entry : entrySet) {
			total += entry.getValue();// 重新计算总数量
		}

		if (way.equals("add")) {// 对数据进行增加操作
			count++;
			total++;
			mobileMap.put(type, count);
			mobileMap.put("total", total);
			return mobileMap;
		}
		throw new RuntimeException("请传入正确的way参数！");
	}

	/**
	 * 根据更定的类型和已经发送过的数量，判断是否还能够发送告警短信<br>
	 * 消息包括：低电告警、断电告警、位移告警（锁车）、震动告警（锁车）<br>
	 * 3—GPS异常断电<br>
	 * 5—GPS低电<br>
	 * 14—GPS震动（上锁）<br>
	 * 16—GPS位移（上锁）<br>
	 * 
	 * @param type
	 *            发送消息的类型
	 * @param count
	 *            已经发送的次数
	 * @return 返回剩余发送短信的数量
	 */
	private int surplusCount(String type, int count) {
		/**
		 * 断电告警（上锁）2次<br>
		 * 震动报警（上锁）3次<br>
		 * 异常位移（上锁）1次<br>
		 */
		switch (type) {
		case "3": {// GPS异常断电
			return 2 - count;
		}
		case "14": {// GPS震动（上锁）
			return 3 - count;
		}
		case "16": {// GPS位移（上锁）
			return 1 - count;
		}
		// case "5": {
		// return false;// 低电量告警补发送短信
		// }
		default:
			return -1;// 其他的不再定义范围内的短信都不发送
		}
	}

	/**
	 * 对mqtt消息进行重复处理
	 * 
	 * @return
	 * @throws IOException
	 */
	private synchronized boolean repetitionFilter(String topic, MqttMessage message) throws IOException {
		String msg = URLDecoder.decode(new String(message.getPayload()), "UTF-8");
		String md5Code = MD5Util.getMD5Code(topic + msg);// 对主题和消息惊醒MD5加密，生成的字符串就是这一条消息的表示
		if (!mqttMsgSet.contains(md5Code)) {// 如果没有接受过这条消息，则进行判断
			JSONObject msgJsonObject = JSON.parseObject(msg);
			String time = msgJsonObject.getString("time");
			Date msgDate = DateUtil.stringToDate(time, DateUtil.DATE_TO_STRING_DETAIAL_PATTERN);
			long sub = System.currentTimeMillis() - msgDate.getTime();
			if (sub < 86400000L) {// 这条消息是否超过24小时
				mqttMsgSet.add(md5Code);// 将这条消息添加到集合中
				String setJsonStr = JSON.toJSONString(mqttMsgSet);
				RWmqttMsgSetFile(setJsonStr, getMqttMsgFile(), "write");// 将集合记录在硬盘中，保证下次服务器启动时读取到的是最新的集合数据
				return true;// 如果未超过24小时，则返回true，即继续处理
			}
		}
		log.debug("这条Mqtt消息是重复的！");
		return false;// 如果接收过这条消息，则返回false，即不处理
	}

	/**
	 * 获取mqtt记录文件
	 * 
	 * @return
	 * @throws IOException
	 */
	public static File getMqttMsgFile() throws IOException {
		String mirkPath = "/usr/local/src/tomcat/project-logs/scps/mqtt-msg-set";// linux下的地址
		// String mirkPath = "d:/project-logs/scps/mqtt-msg-set";// win下的地址
		String fileName = "mqttMsgSet.log";
		File mirk = new File(mirkPath);
		if (!mirk.exists()) {
			mirk.mkdirs();
		}
		File file = new File(mirkPath + "/" + fileName);
		if (!file.exists()) {
			file.createNewFile();
		}
		return file;
	}

	/**
	 * 将mqttMsgSet写入mqtt消息记录文件<br>
	 * 或者<br>
	 * 读取mqtt消息记录文件写入mqttMsgSet集合里<br>
	 * 如果是读取，则会自动初始化mqttMsgSet集合
	 * 
	 * 
	 * @param setJsonStr
	 *            这是需要更新至文件的字符串内容，如果是读取，这个参数可以为null
	 * @param file
	 * @param way
	 *            传入“read”则是读取，其他则是写入,如果是写入操作，则不会对集合进行处理
	 * @throws IOException
	 * @throws FileNotFoundException
	 * @return 如果是读取，则会自动将mqttMsgSet集合初始化
	 */
	public synchronized static void RWmqttMsgSetFile(String setJsonStr, File file, String way)
			throws IOException, FileNotFoundException {
		if ("read".equals(way)) {
			// 读
			BufferedInputStream in = new BufferedInputStream(new FileInputStream(file));
			byte[] b = new byte[1024];
			int len = 0;
			String jsStr = "";
			while ((len = in.read(b)) != -1) {
				jsStr += (new String(b, 0, len));
			}
			if (StringUtils.isNotBlank(jsStr)) {
				JSONArray parseArray = JSON.parseArray(jsStr);
				for (Object object : parseArray) {
					Set<String> set = new HashSet<>();
					set.add((String) object);
					mqttMsgSet = set;
				}
			}
			in.close();
		} else {
			// 写
			if (StringUtils.isEmpty(setJsonStr)) {
				setJsonStr = "";
			}
			BufferedOutputStream out = null;
			try {
				out = new BufferedOutputStream(new FileOutputStream(file, false));
				out.write(setJsonStr.getBytes());
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				if (out != null)
					out.close();
			}
		}
	}

	/**
	 * 发送短信
	 * 
	 * @param jsonObject
	 * @param mobile
	 * @param code
	 * @param surplusCount
	 *            显示剩余多少条的数量
	 */
	protected Map<String, Object> sendSms(JSONObject jsonObject, String mobile, String code, Integer surplusCount) {
		// -------------------------------发送短信逻辑-----------------------------
		Map<String, Object> sendBatchMessage = null;
		ISmsTemplateService smsTemplateService = (ISmsTemplateService) SpringContextUtil
				.getBean("smsTemplateServiceImpl");
		SmsTemplateEntity tplEntity = null;
		String tplCode = code;

		if (StringUtils.isNotBlank(tplCode)) {// 模板code不为空，说明满足了发送短信的要求
			tplEntity = smsTemplateService.findByCode(tplCode);
			SendSmsTools sst = new SendSmsTools();
			String text = "【" + tplEntity.getCompany() + "】" + tplEntity.getContent();
			String time = jsonObject.getString("time");
			String[] split = time.split(" ");
			text = text.replace("#time#", split[1]);
			text = text.replace("#count#", surplusCount + "");
			sendBatchMessage = sst.sendBatchMessage(new String[] { mobile }, text);
			Logger log = Logger.getLogger(getClass());
			log.info("告警短信发送情况:" + sendBatchMessage);
		}
		// -------------------------------发送短信逻辑-----------------------------
		return sendBatchMessage;
	}

	/**
	 * 执行推送，并记录日志
	 * 
	 * @param log
	 * @param messageEntity
	 * @throws IOException
	 * @throws ParseException
	 */
	protected void pushMessage(String mobile, SysMessageEntity messageEntity) throws IOException, ParseException {
		List<String> mobileList = new ArrayList<>();
		mobileList.add(mobile);
		List<String> pagList = new ArrayList<String>();
		String payload = messageEntity.getPayload();
		String description = messageEntity.getPayload();
		String title = messageEntity.getTitle();
		Integer passThrough = 0;
		Integer notifyType = MessageInfo.DEFAULT_ALL;
		MessageInfo messageInfo = new MessageInfo(payload, title, description, passThrough, notifyType, pagList);
		Map<String, String> extra = new HashMap<>();
		// String uri = "";
		// extra.put(Constants.EXTRA_PARAM_NOTIFY_EFFECT,
		// Constants.NOTIFY_ACTIVITY);
		// extra.put(Constants.EXTRA_PARAM_INTENT_URI, uri);
		sysConfigService = (ISysConfigService) SpringContextUtil.getBean("sysConfigServiceImpl");
		String callbackUrl = sysConfigService.findByCode("MIPush_callbackUrl").getValue();
		extra.put("callback", callbackUrl);
		extra.put(Constants.EXTRA_PARAM_NOTIFY_EFFECT, Constants.NOTIFY_LAUNCHER_ACTIVITY);// 设置打开安卓APP
		// extra.put("callback.param", mobile);
		// extra.put("callback.type", "1");
		messageInfo.setExtra(extra);
		MessageFactory androidMessageFactory = MessageFactory.AndroidMessageFactory();// 获取安卓消息工厂
		Message androidMessage = androidMessageFactory.createMessage(messageInfo);// 创建推送消息对象
		ExecuteMIPush miPush = ExecuteMIPush.newInstance();
		Result andRes = miPush.sendToAlias(ExecuteMIPush.androidSender(), androidMessage, mobileList, 3);// 推送安卓消息
		//ios消息
		MessageFactory iosMessageFactory = MessageFactory.IOSMessageFactory();
		Message iosMessage = iosMessageFactory.createMessage(messageInfo);
		Result iosRes = miPush.sendToAlias(ExecuteMIPush.iosSender(), iosMessage, mobileList, 3);// 推送ios消息
		// miPush.sendToUserAccount(createMessage, mobileList, 3);
		Logger log = Logger.getLogger(PushCallback.class);
		log.info("消息已经推送安卓消息，推送的用户：" + mobileList + ";推送内容为：" + payload + ";消息推送结果为：" + andRes);
		log.info("消息已经推送ios消息，推送的用户：" + mobileList + ";推送内容为：" + payload + ";消息推送结果为：" + iosRes);

	}

}
